/*
 Copyright 2009 Abdulla Abdurakhmanov (abdulla@latestbit.com)
 Original sources are available at www.latestbit.com

 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
 You may obtain a copy of the License at

 http://www.apache.org/licenses/LICENSE-2.0

 Unless required by applicable law or agreed to in writing, software
 distributed under the License is distributed on an "AS IS" BASIS,
 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 See the License for the specific language governing permissions and
 limitations under the License.

 TCPAcceptor.cpp. Created on: 20.08.2009 15:57:58
 */

#include "smartobjects/network/tcp/TCPAcceptor.hpp"
#include "smartobjects/network/tcp/TCPAcceptorConnection.hpp"

namespace smartobjects {

const char* TCPAcceptor::ACCEPTOR_MULTIPLEXER_GROUP_NAME = "ACCEPTORS";

TCPAcceptor::TCPAcceptor(TCPSharedResources::SharedPtr poSharedResources):
		m_poSharedResources (poSharedResources),
		m_poSocketURI ( NULL ),
		m_hAcceptorSocket(-1),
		m_bStoppedState(false) {
	m_poMultiplexer = m_poSharedResources->getSocketMultiplexer( ACCEPTOR_MULTIPLEXER_GROUP_NAME );
}

TCPAcceptor::~TCPAcceptor() {
	stop();
	if(m_poSocketURI) {
		m_poSocketURI->release();
		m_poSocketURI = NULL;
	}
}

void TCPAcceptor::addListener(IAcceptorListener* poListener) {
	boost::recursive_mutex::scoped_lock oLock ( m_mxAcceptorListeners );
	if( !m_oPALAcceptorListeners.isActivated() ) {
		m_aAcceptorListeners.push_back(poListener);
	}
	else {
		m_oPALAcceptorListeners.putListenerToAdd(poListener);
	}
}

void TCPAcceptor::delListener(IAcceptorListener* poListener) {
	boost::recursive_mutex::scoped_lock oLock ( m_mxAcceptorListeners );
	if( !m_oPALAcceptorListeners.isActivated() ) {
		m_aAcceptorListeners.remove(poListener);
	}
	else {
		m_oPALAcceptorListeners.putListenerToRemove(poListener);
	}
}

void TCPAcceptor::start(void) throw (const IOException&) {
	boost::recursive_mutex::scoped_lock oLock(m_mxAcceptorSocketLock);
	prepareAcceptorSocket();
	startAcceptorSocketListener();
}

void TCPAcceptor::prepareAcceptorSocket() throw (const IOException&) {
	if(!m_poSocketURI)
		throw SystemException("Acceptor's endpoint URI isn't specified");

	if ( this->isOpen() ) {
		this->stop();
	}

	m_hAcceptorSocket = smartobjects::sockets::osimpl::createSocket(SOCK_STREAM,IPPROTO_TCP);

	if ( m_hAcceptorSocket == -1 ) {
		throw IOException ( "Unable to create an acceptor socket: " + getLastOSErrorString() );
	}

	smartobjects::sockets::osimpl::setReuseAddrOption ( m_hAcceptorSocket, true );

	// Binding address
	int nBindResult = smartobjects::sockets::osimpl::bindAddress ( m_hAcceptorSocket,  m_poSocketURI->getSocketAddr() );

	if(nBindResult == -1 ) {
		smartobjects::sockets::osimpl::close ( m_hAcceptorSocket );
		m_hAcceptorSocket = -1;
		throw IOException ( "Unable to bind the specified address to acceptor socket: " + getLastOSErrorString() );
	}

	smartobjects::sockets::osimpl::setNoneBlockMode ( m_hAcceptorSocket, true );
}

void TCPAcceptor::startAcceptorSocketListener() throw (const IOException&) {
	if ( smartobjects::sockets::osimpl::listenSocket ( m_hAcceptorSocket ) == -1 ) {
		smartobjects::sockets::osimpl::close ( m_hAcceptorSocket );
		m_hAcceptorSocket = -1;
		throw IOException ("Unable to listen on acceptor socket: " + getLastOSErrorString());
	}

	m_bStoppedState = false;

	m_poMultiplexer->addSocketAcceptor (
		m_hAcceptorSocket ,
		boost::bind ( &TCPAcceptor::onAccepted, this )
	);
}


void TCPAcceptor::stop(void) {
	boost::recursive_mutex::scoped_lock oAsyncEventLock ( m_mxAsyncEvent );
	m_bStoppedState = true;
	oAsyncEventLock.unlock();

	resetAllAcceptorConnections();

	boost::recursive_mutex::scoped_lock oLock(m_mxAcceptorSocketLock);
	if ( isOpen() ) {
		m_poMultiplexer->removeSocketAcceptor ( m_hAcceptorSocket );

		smartobjects::sockets::osimpl::shutdown ( m_hAcceptorSocket );
		smartobjects::sockets::osimpl::close ( m_hAcceptorSocket );

		m_hAcceptorSocket = -1;
	}

}

void TCPAcceptor::resetAllAcceptorConnections() {
	boost::recursive_mutex::scoped_lock oLock(m_mxAcceptorConnections);
	AcceptorConnections aTempAcceptorConnections = m_aAcceptorConnections; //Avoiding a deadlock
	m_aAcceptorConnections.clear();
	oLock.unlock();
	for ( AcceptorConnections::iterator it = aTempAcceptorConnections.begin();
						it!= aTempAcceptorConnections.end(); it++) {
			(*it)->close();
			(*it)->release();
	}
	aTempAcceptorConnections.clear();

}

bool TCPAcceptor::isOpen() {
	boost::recursive_mutex::scoped_lock oLock(m_mxAcceptorSocketLock);
	return (this->m_hAcceptorSocket != -1);
}

void TCPAcceptor::setEndpointURI(URI* poURI) throw (const IOException&) {
	const std::string& sURI = poURI->getURIStr();
	if(m_poSocketURI) {
		m_poSocketURI->release();
		m_poSocketURI = NULL;
	}
	m_poSocketURI = new smartobjects::sockets::SocketURI( sURI );
}

URI* TCPAcceptor::getEndpointURI(void) {
	return m_poSocketURI;
}

void TCPAcceptor::onAccepted() {
	boost::recursive_mutex::scoped_lock oAsyncEventLock ( m_mxAsyncEvent );
	if(this->m_bStoppedState)
		return;

	boost::recursive_mutex::scoped_lock oLock(m_mxAcceptorSocketLock);
	smartobjects::sockets::SocketAddr oClientAddr;
	smartobjects::sockets::Socket hClientSocket = -1;
	if(isOpen())
		hClientSocket = smartobjects::sockets::osimpl::acceptClient (m_hAcceptorSocket, oClientAddr);
	oLock.unlock();

	if ( hClientSocket !=-1 ) {
		SafeForReleaseMethodLock oReleaseLock(this);

		TCPAcceptorConnection* poConnection
			= createAcceptorConnection(hClientSocket, oClientAddr);

		m_mxAcceptorConnections.lock();
		m_aAcceptorConnections.insert( poConnection );
		m_mxAcceptorConnections.unlock();

		if(poConnection->getHandshakeManager()==NULL) {
			// If there is no handshake procedure
			fireOnAccepted(poConnection);
		}

		try {
			poConnection->start();
		}
		catch(const std::exception& ex) {
			// FIXME: Debug prints
			std::cout << "Internal exception: " << ex.what() << std::endl;
		}
	}
	else {
		// fireOnAcceptFailure ( getLastOSErrorString() );
	}

}

TCPAcceptorConnection* TCPAcceptor::createAcceptorConnection( smartobjects::sockets::Socket hClientSocket, smartobjects::sockets::SocketAddr& oClientAddr ) {
	TCPAcceptorConnection* poConnection = new TCPAcceptorConnection( m_poSharedResources, this );

	poConnection->setSocket( hClientSocket );

	// Setup connection endpoint as remote endpoint
	boost::intrusive_ptr< smartobjects::sockets::SocketURI > poTCPEndpointURI (
			new smartobjects::sockets::SocketURI ( oClientAddr, m_poSocketURI->getScheme()),
			false
	);
	poConnection->setEndpointURI( poTCPEndpointURI.get() );

	return poConnection;
}


void TCPAcceptor::fireOnAccepted(IConnection* poConnection) {
	boost::recursive_mutex::scoped_lock oLock(m_mxAcceptorListeners);
	m_oPALAcceptorListeners.activate();
	foreach(IAcceptorListener* poListener, m_aAcceptorListeners) {

		poListener->onAccepted(poConnection);
	}
	processingPostActionAcceptorListeners();
}


void TCPAcceptor::processingPostActionAcceptorListeners() {
	m_oPALAcceptorListeners.deactivate();
	foreach( PostActionListenersContainer<IAcceptorListener>::PostActionQueueItem item, m_oPALAcceptorListeners.getPostActionQueue()) {
		if(item.isActionToAdd()) {
			addListener( item.getListener() );
		}
		else
		if(item.isActionToRemove()) {
			delListener( item.getListener() );
		}
	}
	m_oPALAcceptorListeners.reset();
}

void TCPAcceptor::removeConnection(TCPAcceptorConnection* poConnection) {
	m_mxAcceptorConnections.lock();
	if(m_aAcceptorConnections.find(poConnection)!=m_aAcceptorConnections.end()) {
		m_aAcceptorConnections.erase ( poConnection );
		poConnection->release();
	}
	m_mxAcceptorConnections.unlock();
}

}
